package hyl.ext.pipe;

import java.util.HashMap;
import java.util.Map;

import org.apache.rocketmq.client.exception.MQClientException;

import hyl.base.mq.rocket.MyRocketMq;
import hyl.core.MyFun;
import hyl.core.info.CMsg;
import hyl.core.io.MyPath;
import hyl.core.reflect.MyRef2;
import hyl.core.run.IFinishT;
import hyl.core.run.MyRun;

public class CCFrame {
	CCConfig __配置 = null;
	String s模型名称 = null;
	boolean b是否禁用 = false;
	String s包文件 = null;
	Class<?> _c主类 = null;
	Object _类对象 = null;
	MyRocketMq _mqp = null;
	// MyRocketMq[] _mqcs = null;
	MyRocketMq _mqc = null;
	String mq_消费配置 = null;
	String mq_生产配置 = null;
	// int mq_消费者_MAX = 1;
	Object o处理对象 = null;
	// 配置时注入所有配置节点
	Map<String, CCNode> _节点集 = new HashMap<>();

	public CCFrame(CCConfig 配置, String 生产配置, String 消费配置, String 模型, String 包文件, String 主类, boolean 是否禁用)
			throws Exception {
		__配置 = 配置;
		mq_生产配置 = 生产配置;
		mq_消费配置 = 消费配置;
		// mq_消费者_MAX = 消费者_MAX;

		s模型名称 = 模型;
		s包文件 = 包文件;
		b是否禁用 = 是否禁用;
		if (MyFun.isEmpty(s包文件) || s包文件.length() == 1) {
			_c主类 = Class.forName(主类);
		} else {
			String path = (s包文件.startsWith(".")) ? MyPath.getSystemResource(s包文件) : s包文件;
			_c主类 = MyRef2.getJarClass(path, 主类);
		}
		o处理对象 = _c主类.newInstance();
	}

	public void addNode(CCNode wk) {
		_节点集.put(wk.s来源, wk);
	}

	public void close() {
		_mqp.close();
		// for (MyRocketMq mq : _mqcs)
		// mq.close();
		_mqc.close();
	}

	MyRun _线程池 = null;

	FrameCallback cb2 = new FrameCallback();

	public MyRocketMq startProducer() throws MQClientException {
		return startProducer(null);
	}

	public MyRocketMq startProducer(IFinishT<CMsg> callback) throws MQClientException {
		System.out.println("开启生产者...");
		if (_mqp == null) {
			_mqp = MyRocketMq.iniProducer(mq_生产配置, "p" + s模型名称, 0);
			_mqp.ready(s模型名称, callback);
			_mqp.startProducer();
		}
		return _mqp;
	}

	public MyRocketMq startConsumer() {

		System.out.println("开启消费者...");
//		if (_mqcs == null) {
//			_mqcs = new MyRocketMq[mq_消费者_MAX];
//			// 启动所有消费者并行消费
//			for (int i = 0; i < mq_消费者_MAX; i++) {
//				//System.out.println("消费id=" + i);
//				_mqcs[i] = MyRocketMq.iniConsumer(mq_消费配置, "c" + s模型名称, i);
//				_mqcs[i].ready(s模型名称, cb2);
//			}
//			// 启动多个消费者 处理消息
//			for (MyRocketMq mq : _mqcs) {
//				//System.out.println("接收" + mq);
//				mq.receive();
//			}
//		}
		_mqc = MyRocketMq.iniConsumer(mq_消费配置, "c" + s模型名称, 0);
		_mqc.ready(s模型名称, cb2);
		_mqc.receive();
		return _mqc;
	}

	public void start() throws MQClientException {
		if (b是否禁用)
			return;

		startProducer();
		startConsumer();

	}

	public void v停止() {
		if (b是否禁用)
			return;
		MyRun.close(s模型名称);
	}

	public void send(String 去向, byte[] 内容) {
		_mqp.send(去向, 内容);
	}

	/**
	 * 节点接收消息, 交给回调函数统一处理,处理完毕, 重新把消息发出去<br>
	 * 在处理过程消息的接收处理 与线程
	 * 
	 * @author 37798955@qq.com
	 *
	 */
	class FrameCallback implements IFinishT<CMsg> {
		// 这里的返回值其实没有意义,因为后台不处理
		@Override
		public boolean success(CMsg o) {
			// 选择对应的节点
			// 分配线程去运行 ,运行结束收回线程,并且手动提交处理完成的
			// 如果处理失败 发送会原消息队列即可
			// System.out.println("收到消息处理---"+o.toJsonString());
			if (_节点集.containsKey(o.s键)) {
				CCNode nd = _节点集.get(o.s键);
				try {
					// System.out.println("接收到:" + o._skey);
					Boolean result = (Boolean) nd._方法.invoke(o处理对象, o);
					if (result == null || result) {
						if (nd.a去向 != null) {
							// System.out.println("从" + o._skey + ",到" + MyFun.array2Str(nd.a去向, ","));
							for (String key : nd.a去向) {
								_mqp.send(key, o.bs内容);
							}
						}
					}
				} catch (Exception e) {
					e.printStackTrace();
					// System.out.println("异常后 重新发到" + o._skey);
					_mqp.send(o.s键, o.bs内容);
				}
			}
			return true;
		}

		@Override
		public void error(CMsg o, Throwable e) {
			e.printStackTrace();
			if (o != null)
				_mqp.send(o.s键, o.bs内容);
		}

	}

}
